#pragma once

#ifndef ORCL_H
#define ORCL_H

#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <numaif.h>
#include <thread>
#include <malloc.h>
#include <time.h>
#include <fstream>
#include <iostream>
#include <infiniband/verbs.h>
#include <infiniband/verbs_exp.h>
#include <libmemcached/memcached.h>
#include "ORCL_size.h"
#include "extent_param.h"

using namespace std;

#define ORCL_DCT_KEY 3185

#define ORCL_SQ_DEPTH 1024	/* Depth of all SEND queues */
#define ORCL_RQ_DEPTH 4096	/* Depth of all RECV queues */

#define ORCL_DEFAULT_PSN 3185	/* PSN for all queues */
#define ORCL_DEFAULT_QKEY 0x11111111
#define ORCL_MAX_LID 256

#define ORCL_QP_NAME_SIZE 200	/* Size (in bytes) of a queue pair name */
#define ORCL_RESERVED_NAME_PREFIX "__ORCL_RESERVED_NAME_PREFIX"

#define ORCL_RC_MAX_INLINE 208
#define ORCL_DCT_MAX_INLINE 171

// 每个组中最大支持的节点数量
#define ORCL_MAX_GROUP_SIZE 16
// 最大组的数量
#define ORCL_MAX_GROUP_NUM 16
// 每个上下文最大的DCI数量
#define ORCL_MAX_DCI_NUM 16
// 最大的节点数量
#define ORCL_MAX_NODE_NUM 64


// 转发请求接收缓冲区的大小
#define FORWARD_DCT_BUF_SIZE 67108864
#define FORWARD_RC_BUF_SIZE 67108864
#define DCI_DM_SIZE 16384


#define ORCL_MOD_ADD(x, N) do { \
	x = x + 1; \
	if(x == N) { \
		x = 0; \
	} \
} while(0)

// 返回当中的小值和大值
#define ORCL_MIN(a, b) (a < b ? a : b)
#define ORCL_MAX(a, b) (a > b ? a : b)

#define CPE(val, msg, err_code) \
	if(val) { fprintf(stderr, msg); fprintf(stderr, " Error %d \n", err_code); \
	exit(err_code);}

/* Ensure that x is between a and b, inclusive */
#define range_assert(x, a, b) (assert(x >= a && x <= b))

/* Compile time assert. !!(@condition) converts @condition into a 0/1 bool. */
#define ct_assert(condition) ((void) sizeof(char[-1 + 2 * !!(condition)]))

#define likely(x)      __builtin_expect(!!(x), 1)
#define unlikely(x)    __builtin_expect(!!(x), 0)
#define forceinline inline __attribute__((always_inline))
#define _unused(x) ((void)(x))	/* Make production build happy */

/* Is pointer x aligned to A-byte alignment? */
#define is_aligned(x, A) (((uint64_t) x) % A == 0)

/* Registry info about a QP */
struct orcl_qp_attr {
	char name[ORCL_QP_NAME_SIZE];

	/* Info about the RDMA buffer associated with this QP */
	volatile uint8_t* buf_addr;
	uint32_t buf_size;
	uint32_t rkey;
	
	int lid;
	int qpn;

    // 用gid来进行传输
    union ibv_gid gid;
};

struct orcl_dct_attr {
    char name[ORCL_QP_NAME_SIZE];

	// DCT target对应的本地id
	int lid;
	// 对应lid的地址下，dct target的号码
	int dct_num;

	/* Info about the RDMA buffer associated with this QP */
	volatile uint8_t* buf_addr;
    uint32_t buf_size;
	uint32_t rkey;

    // 使用gid来进行传输
    union ibv_gid gid;
};

// // 转发线程要传入的参数
// struct orcl_forward_param{
// 	// 其监听的缓冲区
// 	volatile uint8_t* dct_buf;
// 	// 转发发送缓冲区的位置
// 	volatile uint8_t *rc_forward_buf;

// 	// 监听缓冲区的位置，以及将数据放到发送缓冲区的位置
// 	volatile int* forward_offset;

// };

// orcl上下文
struct orcl_context{
    // 当前节点在整个集群中的ID号，也是这个上下文的id号。我们假设一个节点一个上下文，虽然实际上并不是这样
    int node_id;
    // 当前节点的组号
    int group_id;

    // 节点数量
    int node_num;
    // 组的数量
    int group_num;
    // 当前组节点的数量，rc的数量是这个数-1
    int group_size;

    // 当前上下文dci的数量
    int dci_num;

    // 因为wr的提交是按照ctx上锁的，所以两组RC和一个DCI分别申请三种ctx
    // 因为同一个组不用建立DCT，同一个节点也不用建立RC，实际上不需要那么多
    struct ibv_context *rc_normal_ctx[ORCL_MAX_GROUP_SIZE];
    struct ibv_context *rc_forward_ctx[ORCL_MAX_GROUP_SIZE];
    
    // 用一个数组来保存同组的其他rc信息
    struct orcl_qp_attr others_normal_qp[ORCL_MAX_GROUP_SIZE];
    struct orcl_qp_attr others_forward_qp[ORCL_MAX_GROUP_SIZE];
	struct orcl_dct_attr others_dct[ORCL_MAX_GROUP_NUM];
	// 与上面一个数据结构配套，远程节点的路由
	// 这个路由每个dci独一份，所以是个二维数组
	struct ibv_ah* other_dct_ah_attr[ORCL_MAX_DCI_NUM][ORCL_MAX_GROUP_NUM];

    // dct用来组间通信
    struct ibv_context *dci_ctx[ORCL_MAX_DCI_NUM];
    // 用一个数组保存当前DCT连接的组
    int dci_connect_group_id[ORCL_MAX_DCI_NUM];
    // 用一个数组保存每个组使用的DCT
    int group_connect_dci_id[ORCL_MAX_GROUP_NUM];
    // 使用panda的算法，用一个编号来声明要负责新连接DCI编号
    int next_dci_id;

    // dct和dci不能在一起，需要一系列的元数据来维护dct
    struct ibv_context *dct_ctx[ORCL_MAX_GROUP_NUM];

    // 端口号，我们只支持单端口网卡
	int port_index;	/* User-supplied. 0-based across all devices */
	// 设备号
	int device_id;	/* Resovled by liborcl from @port_index */
	// 设备端口id
	int dev_port_id;	/* 1-based within dev @device_id. Resolved by liborcl */
	// numa中的节点id，也就是不同的核
	int numa_node_id;	/* NUMA node id */


    // 保护域，针对三种上下文分别有不同数量和种类的保护域
	struct ibv_pd *rc_normal_pd[ORCL_MAX_GROUP_SIZE];
    // normal只会读写message_buf，forward除了message_buf，还有转发缓冲区
    struct ibv_pd *rc_forward_pd[ORCL_MAX_GROUP_SIZE];
    struct ibv_pd *dci_pd[ORCL_MAX_DCI_NUM];
    struct ibv_pd *dct_pd[ORCL_MAX_GROUP_NUM];

    // 针对三种类型的队列，分别设计完成队列
    // 自己连接和同组连接的位置置空就行
	struct ibv_qp *rc_forward_qp[ORCL_MAX_GROUP_SIZE];
	struct ibv_cq *rc_forward_cq[ORCL_MAX_GROUP_SIZE];
    struct ibv_qp *rc_normal_qp[ORCL_MAX_GROUP_SIZE];
	struct ibv_cq *rc_normal_cq[ORCL_MAX_GROUP_SIZE];
    
    // 需要一个数组记录每个rc在当前的ctx下传输的次数，用来周期signal
    int rc_forward_signal_period[ORCL_MAX_GROUP_SIZE];
    int rc_normal_signal_period[ORCL_MAX_GROUP_SIZE];
    int dci_signal_period[ORCL_MAX_DCI_NUM];
    // 关于signal周期的记录是独享的

    struct ibv_qp *dci_qp[ORCL_MAX_DCI_NUM];
    struct ibv_cq *dci_cq[ORCL_MAX_DCI_NUM];

	// 每一个dci有一个dm缓冲区
	struct ibv_exp_dm *dci_dm[ORCL_MAX_DCI_NUM];
	// 每个缓冲区有自己的mr
	struct ibv_mr *dci_dm_mr[ORCL_MAX_DCI_NUM];
	// 用一个变量存储当前dm要被写入的位置
	int dm_write_offset[ORCL_MAX_DCI_NUM];

    // 每个上下文的dct数量为组的数量-1，每个dct负责一个组的转发，每个dct用一个额外的线程来负责
    struct ibv_exp_dct *dc_target[ORCL_MAX_GROUP_NUM];
    struct ibv_srq *dc_target_srq[ORCL_MAX_GROUP_NUM];
    struct ibv_cq *dc_target_cq[ORCL_MAX_GROUP_NUM];
    // 每个上下文使用的线程实例，每个线程负责一个dct的接收与转发
    std::thread thread_arr[ORCL_MAX_GROUP_NUM];
    // 对用户不透明的转发（环形）缓冲区，分为两个部分，一个是传入的dct缓冲区，一个是传出缓冲区，为了方便申请大页，也要配置成共享的
    // dct_forward_buf是发送完就复原的，RC不服原，新值覆盖就可以
    volatile uint8_t* dct_forward_buf[ORCL_MAX_GROUP_NUM];// 每一个缓冲区给一个线程和一个dct
    // 需要用一个数组记录一下要写远程DCT缓冲区的位置，对于RC缓冲区的发送偏移量，
    volatile int forward_offset[ORCL_MAX_GROUP_NUM];
	// 当前缓冲区的监听位置，可以放在线程做临时变量
	// volatile int listen_offset[ORCL_MAX_GROUP_SIZE];
	

    volatile struct ibv_mr* dct_forward_mr[ORCL_MAX_GROUP_NUM];
    // 每个线程使用rc来转发数，这个数组里面的一个缓冲区给一个dct和一个线程
    volatile uint8_t *rc_forward_buf[ORCL_MAX_GROUP_NUM];
    volatile struct ibv_mr* rc_forward_mr[ORCL_MAX_GROUP_NUM][ORCL_MAX_GROUP_SIZE];
    // 上面这一组从dc target到rc_forward_buf这一串东西，为一个转发线程独享

    // 对用户透明的发送缓冲区（不是环形的，各种方式访问，并且是共享的）
    volatile uint8_t *message_buffer;
    int message_buffer_size;
    // 所有类型的队列都需要对这个缓冲区进行注册
    volatile struct ibv_mr* dci_mes_mr[ORCL_MAX_DCI_NUM];
    volatile struct ibv_mr* rc_normal_mes_mr[ORCL_MAX_GROUP_SIZE];
    volatile struct ibv_mr* rc_forward_mes_mr[ORCL_MAX_GROUP_SIZE];

    // 整个数据流是这样的
    // message_buffer-网络传输->dct_forward_buf-内存拷贝->rc_forward_buf-网络传输->message_buffer

    // 每个ctx是不支持多线程的，但是我们可以通过批量申请来让同一批上下文共享rc的元数据，但是dct不是共享的，message_buffer不是共享的
    // 批量申请只能进行一次，并且申请的每一个ctx要用一个东西加以区分
    int orcl_ctx_id;

    // 规定当前其实shmkey，在这个项目中，打大量缓冲区需要用共享内存的方式来申请大页
    // 主要包括message_buffer 1个，rc转发缓冲区group size个，dct转发区group number个，我们在shm_start的基础上增加就好了
    // 加到的位置记录在shm_key_end中
    int shm_key_start;
    int shm_key_end;
};

struct forward_thread_params{
	struct orcl_context* orcl_ctx;
	int tid;
};

// 数据包头部，包含发送的终点，写终点的位置，大小
struct orcl_message_header{
	char magic;
	char node_id;
	int offset;

	// size是带上头部的总大小
	short size;
};

// 我们首先先进行上面这个程序的初始化工作，如果
struct orcl_context* orcl_context_init(int node_id, int group_id, int node_num, int group_num, int group_size, int dci_num, int port_index, 
    int numa_node_id, volatile void *prealloc_mes_buf, int mes_buf_size, int shm_key_start, volatile orcl_context* pre_ctx, int orcl_ctx_id);

// 最终提供的发送函数，此函数提供一个RDMA_write操作
// 传入参数有两个，一个是上下文，一个是message的offset
// 这个offset是要发送的包的首地址，里面自带了包头
// 我们认为offset也是对齐的
void orcl_write_message_to_remote(struct orcl_context* orcl_ctx, int offset);

// 遍历机器上所有网卡的所有端口，并按照0-base编号，然后选择port_index号端口
struct ibv_device* orcl_resolve_port_index(struct orcl_context* orcl_ctx, int port_index);

// 创建第i个normal rc
void orcl_create_normal_conn_qps(struct orcl_context *orcl_ctx, int i);

// 创建第i个forward rc
void orcl_create_forward_conn_qps(struct orcl_context *orcl_ctx, int i);

// 对对应的numa节点申请大页共享缓冲区
void* orcl_malloc_socket(int shm_key, int size, int socket_id);

void* orcl_malloc_socket_no_hugepage(int size, int socket_id);

// 打印
void orcl_red_printf(const char *format, ...);

// 发布全部rc队列的信息，包括normal和forward
void orcl_publish_conn_qp(struct orcl_context* ctx);

// 发布DCT的基本信息
void orcl_publish_dc_target(struct orcl_context* ctx);

// 获取所有当前上下文需要的dct
void orcl_get_all_dc_target(struct orcl_context* orcl_ctx);

// 获取远程节点的qp信息
struct orcl_qp_attr* orcl_get_published_qp(const char *qp_name);

// 获取远程节点的dct信息
struct orcl_dct_attr* orcl_get_published_dct(const char *dct_name);

// 获取远程节点信息
int orcl_get_published(const char *key, void **value);

// 根据上下文和端口号，找出lid
uint16_t orcl_get_local_lid(struct ibv_context *ctx, int dev_port_id);

// 向全局共享内存共享消息
void orcl_publish(const char *key, void *value, int len);

// 获取环境变量
char* orcl_getenv(const char *name);

// 运行向缓冲区的数据拷贝
// 参数首先是上下文，源起始位置，目的dm的编号，目的偏移量，拷贝大小
void cpy_to_device(struct orcl_context* orcl_ctx, uint8_t* source_buf, int source_offset, int dm_index, int dest_offset, int size);

// 创造一个memcache实例
memcached_st* orcl_create_memc();

// 所有的RC节点建立连接
void orcl_all_qp_connect(struct orcl_context* ctx);

// 让一个QP与远程建立连接
void orcl_normal_qp_connect(struct orcl_context* ctx, int i, struct orcl_qp_attr *remote_qp_attr);
void orcl_forward_qp_connect(struct orcl_context* ctx, int i, struct orcl_qp_attr *remote_qp_attr);

// 组内两个节点互相通信，参数是目标节点的组内index
// 发送位置在messagebuf中的offset，接收位置的offset
// 已经要传输的size大小
void orcl_inner_group_rdma_write(struct orcl_context* orcl_ctx, int dest_inner_group_index, int source_offset, int dest_offset, int size);

// 组内两个节点的通信，使用forward rc在组内进行转发
// 从rc的发送缓冲区发往远程的message buf
// 需要一个额外的参数来标明发送源来自于哪一个forward缓冲区
void orcl_inner_group_rdma_forward(struct orcl_context* orcl_ctx, int forward_buf_index, int dest_inner_group_index, int source_offset, int dest_offset, int size);

// 向转发节点的对应缓冲区发送消息，消息512对齐
void orcl_dci_rdma_write(struct orcl_context* orcl_ctx, int dest_group_index, int source_offset, int dest_offset, int size);

// 等待一个cq完成
/* Fill @wc with @num_comps comps from this @cq. Exit on error. */
static inline void
orcl_poll_cq(struct ibv_cq *cq, int num_comps, struct ibv_wc *wc)
{
	int comps = 0;

	while(comps < num_comps) {
		int new_comps = ibv_poll_cq(cq, num_comps - comps, &wc[comps]);
		if(new_comps != 0) {

			/* Ideally, we should check from comps -> new_comps - 1 */
			if(wc[comps].status != 0) {
				fprintf(stderr, "Bad wc status %d\n", wc[comps].status);
				exit(0);
			}

			comps += new_comps;
		}
	}
}

/* Utility functions */
// 快速产生一个随机值
static inline uint32_t orcl_fastrand(uint64_t *seed)
{
    *seed = *seed * 1103515245 + 12345;
    return (uint32_t) (*seed >> 32);
}

// 获取所有的normal和forward远程qp的内容
void get_all_published_qp(struct orcl_context* orcl_ctx);

// 下面这两个函数
// 等待某一个关键字ready
void orcl_wait_till_ready(const char *key_str);

// 发布关键字的ready
void orcl_publish_ready(const char *key_str);

// 创建第i个dci
void orcl_create_dc_initiators(struct orcl_context* orcl_ctx, int i);

// 创建第i个dct
void orcl_create_dc_target(struct orcl_context* orcl_ctx, int i);

// 检查DCT支持，主要是检查设备
static inline void check_dct_supported(struct ibv_context *ctx)
{
	printf("Checking if DCT is supported.. ");
	// 设备参数
	struct ibv_exp_device_attr dattr;
	
	dattr.comp_mask = IBV_EXP_DEVICE_ATTR_EXP_CAP_FLAGS |
					  IBV_EXP_DEVICE_DC_RD_REQ | IBV_EXP_DEVICE_DC_RD_RES;
	int err = ibv_exp_query_device(ctx, &dattr);
	if (err) {
		printf("couldn't query device extended attributes\n");
		assert(false);
	} else {
		if (!(dattr.comp_mask & IBV_EXP_DEVICE_ATTR_EXP_CAP_FLAGS)) {
			printf("no extended capability flags\n");
			assert(false);
		}
		if (!(dattr.exp_device_cap_flags & IBV_EXP_DEVICE_DC_TRANSPORT)) {
			printf("DC transport not enabled\n");
			assert(false);
		}

		if (!(dattr.comp_mask & IBV_EXP_DEVICE_DC_RD_REQ)) {
			printf("no report on max requestor rdma/atomic resources\n");
			assert(false);
		}

		if (!(dattr.comp_mask & IBV_EXP_DEVICE_DC_RD_RES)) {
			printf("no report on max responder rdma/atomic resources\n");
			assert(false);
		}
	}

	printf("Success\n");
}

// 检查设备的dct参数
static inline void check_dct_healthy(struct ibv_exp_dct *dct)
{
	struct ibv_exp_dct_attr dcqattr;
	dcqattr.comp_mask = 0;

	int err = ibv_exp_query_dct(dct, &dcqattr);
	if (err) {
		printf("query dct failed\n");
		assert(false);
	} else if (dcqattr.dc_key != ORCL_DCT_KEY) {
		printf("queried dckry (0x%llx) is different then provided at "
			   "create (0x%llx)\n",
			   (unsigned long long) dcqattr.dc_key,
			   (unsigned long long) ORCL_DCT_KEY);
		assert(false);
	} else if (dcqattr.state != IBV_EXP_DCT_STATE_ACTIVE) {
		printf("state is not active %d\n", dcqattr.state);
		assert(false);
	}
}

// 运行转发函数
void run_forward_thread(struct forward_thread_params *params);

// 另外一套上下文，用来执行延迟优先的程序，与之前提供异步发送不同，这套上下文使用的是同步发送
// 用的是dci+dm的方法
// 结构体中的数据是前者的子集，去掉rc和转发的部分
// 每个上下文只保留一个dci
struct orcl_context_sync{
	// 当前节点在整个集群中的ID号，也是这个上下文的id号。我们假设一个节点一个上下文，虽然实际上并不是这样
    int node_id;
    // 节点数量
    int node_num;

    // 当前上下文dci的数量
    int dci_num;

	struct orcl_dct_attr others_dct[ORCL_MAX_NODE_NUM];
	// 与上面一个数据结构配套，远程节点的路由
	// 这个路由每个dci独一份，所以是个二维数组
	struct ibv_ah* other_dct_ah_attr[ORCL_MAX_DCI_NUM][ORCL_MAX_NODE_NUM];

    // dct用来组间通信
    struct ibv_context *dci_ctx[ORCL_MAX_DCI_NUM];
    // 用一个数组保存当前DCT连接的node
    int dci_connect_node_id[ORCL_MAX_DCI_NUM];
    // 用一个数组保存每个node使用的DCT
    int node_connect_dci_id[ORCL_MAX_NODE_NUM];
    // 使用panda的算法，用一个编号来声明要负责新连接DCI编号
    int next_dci_id;

    // dct和dci不能在一起
    struct ibv_context *dct_ctx;

    // 端口号，我们只支持单端口网卡
	int port_index;	/* User-supplied. 0-based across all devices */
	// 设备号
	int device_id;	/* Resovled by liborcl from @port_index */
	// 设备端口id
	int dev_port_id;	/* 1-based within dev @device_id. Resolved by liborcl */
	// numa中的节点id，也就是不同的核
	int numa_node_id;	/* NUMA node id */


    // 保护域，针对三种上下文分别有不同数量和种类的保护域
    struct ibv_pd *dci_pd[ORCL_MAX_DCI_NUM];
    struct ibv_pd *dct_pd;

    int dci_signal_period[ORCL_MAX_DCI_NUM];
    // 关于signal周期的记录是独享的

    struct ibv_qp *dci_qp[ORCL_MAX_DCI_NUM];
    struct ibv_cq *dci_cq[ORCL_MAX_DCI_NUM];

	// 每一个dci有一个dm缓冲区
	struct ibv_exp_dm *dci_dm[ORCL_MAX_DCI_NUM];
	// 每个缓冲区有自己的mr
	struct ibv_mr *dci_dm_mr[ORCL_MAX_DCI_NUM];
	// 用一个变量存储当前dm要被写入的位置
	int dm_write_offset[ORCL_MAX_DCI_NUM];

    // 每个上下文只有一个dct
    struct ibv_exp_dct *dc_target;
    struct ibv_srq *dc_target_srq;
    struct ibv_cq *dc_target_cq;

    // 对用户透明的发送缓冲区（不是环形的，各种方式访问，并且是共享的）
    volatile uint8_t *message_buffer;
    int message_buffer_size;
    // 所有类型的队列都需要对这个缓冲区进行注册
    volatile struct ibv_mr* dci_mes_mr[ORCL_MAX_DCI_NUM];
    // 接收dct也要对message缓冲区进行注册
    volatile struct ibv_mr* dct_mes_mr;

    // 每个ctx是不支持多线程的，但是我们可以通过批量申请来让同一批上下文共享rc的元数据，但是dct不是共享的，message_buffer不是共享的
    int orcl_ctx_id;

    // 规定当前其实shmkey，在这个项目中，打大量缓冲区需要用共享内存的方式来申请大页
    // 主要包括message_buffer 1个，rc转发缓冲区group size个，dct转发区group number个，我们在shm_start的基础上增加就好了
    // 加到的位置记录在shm_key_end中
    int shm_key_start;
    int shm_key_end;
};

// 初始化orcl_context_sync
struct orcl_context_sync* orcl_context_sync_init(int node_id, int node_num, int dci_num, int port_index, int numa_node_id, volatile void *prealloc_mes_buf, int mes_buf_size, int shm_key_start, int orcl_ctx_id);

struct ibv_device* sync_resolve_port_index(struct orcl_context_sync* sync_ctx, int port_index);

void sync_create_dc_initiators(struct orcl_context_sync* sync_ctx, int i);

void sync_create_dc_target(struct orcl_context_sync* sync_ctx);

void sync_publish_dc_target(struct orcl_context_sync* sync_ctx);

void sync_get_all_dc_target(struct orcl_context_sync* sync_ctx);

// 发送数据，不需要包头，需要包含上下文，源offset，远程节点id，远程offset，要拷贝的size
void sync_write_message_to_remote(struct orcl_context_sync* sync_ctx, int offset, int remote_node_id,int remote_offset, int size);

// 接收数据，不需要包头
void sync_read_message_from_remote(struct orcl_context_sync* sync_ctx, int offset, int remote_node_id,int remote_offset, int size);

void sync_cpy_to_device(struct orcl_context_sync* sync_ctx, uint8_t* source_buf, int source_offset, int dm_index, int dest_offset, int size);

void APPEND_2_LOG(char* file_name, char* append_string);

#endif